package com.newhis.config;

import cn.hutool.core.thread.NamedThreadFactory;
import com.alibaba.nacos.api.config.ConfigService;
import com.alibaba.nacos.api.config.listener.Listener;
import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.spring.factory.NacosServiceFactory;
import com.alibaba.nacos.spring.util.NacosBeanUtils;
import com.alibaba.nacos.spring.util.NacosUtils;
import com.newhis.thread.DyLinkedBlockingQueue;
import com.newhis.thread.ThreadPoolExecutorMdcWrapper;
import org.apache.commons.collections4.MapUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.AsyncConfigurer;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.scheduling.support.CronTrigger;

import java.util.Collection;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Configuration
public class ThreadPoolConfig implements AsyncConfigurer {
    private static final Logger log = LoggerFactory.getLogger(ThreadPoolConfig.class);
    @Value("${thread.pool.coreSize}")
    private int coreSize = 10;
    @Value("${thread.pool.maxSize}")
    private int maxSize = 100;
    @Value("${thread.pool.queueSize}")
    private int queueSize = 50;
    @Value("${thread.pool.keepalive}")
    private int keepalive = 2;
    @Value("${thread.pool.threadName}")
    private String threadName = "hisThreadPoolTaskExecutor-";
    @Value("${nacos.config.thread.dataId}")
    private String threadDataId;
    @Value("${nacos.config.thread.group}")
    private String threadGroup;

    private volatile ThreadPoolExecutorMdcWrapper taskExecutor = null;

    @Bean(name = "threadPoolTaskExecutor")
    public ThreadPoolTaskExecutor newAsync() {
        taskExecutor = new ThreadPoolExecutorMdcWrapper();
        //设置核心线程数
        taskExecutor.setCorePoolSize(coreSize);
        // 线程池维护线程的最大数量，只有在缓冲队列满了以后才会申请超过核心线程数的线程
        taskExecutor.setMaxPoolSize(maxSize);
        //缓存队列
        taskExecutor.setQueueCapacity(queueSize);
        //允许的空闲时间，当超过了核心线程数之外的线程在空闲时间到达之后会被销毁
        taskExecutor.setKeepAliveSeconds(keepalive);
        //异步方法内部线程名称
        taskExecutor.setThreadNamePrefix(threadName);
        //拒绝策略
        taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        taskExecutor.initialize();

        log.info("初始化线程池，核心线程数:{},线程的最大数量：{},缓存队列:{},空闲时间:{}ms,线程名称:{}", taskExecutor.getCorePoolSize(), taskExecutor.getMaxPoolSize(), taskExecutor.getQueueCapacity(), taskExecutor.getKeepAliveSeconds(), taskExecutor.getThreadNamePrefix());
        return taskExecutor;
    }

    @Override
    public Executor getAsyncExecutor() {
        return this.newAsync();
    }

    @Bean
    public void refreshThreadConfig() throws NacosException {
        NacosServiceFactory nacosServiceFactoryBean = NacosBeanUtils.getNacosServiceFactoryBean();
        Collection<ConfigService> configServices = nacosServiceFactoryBean.getConfigServices();
        if (configServices != null && !configServices.isEmpty()){
            for (ConfigService configService : configServices){
                configService.addListener(threadDataId, threadGroup, new Listener() {
                    @Override
                    public Executor getExecutor() {
                        return taskExecutor;
                    }

                    @Override
                    public void receiveConfigInfo(String configInfo) {
                        Map<String, Object> properties = NacosUtils.toProperties(configInfo);
                        if (taskExecutor != null){
                            ThreadPoolExecutor threadPoolExecutor = taskExecutor.getThreadPoolExecutor();
                            //设置核心线程数
                            threadPoolExecutor.setCorePoolSize(MapUtils.getInteger(properties, "thread.pool.coreSize", coreSize));
                            // 线程池维护线程的最大数量，只有在缓冲队列满了以后才会申请超过核心线程数的线程
                            threadPoolExecutor.setMaximumPoolSize(MapUtils.getInteger(properties, "thread.pool.maxSize", maxSize));
                            //缓存队列
                            DyLinkedBlockingQueue executorQueue = (DyLinkedBlockingQueue)threadPoolExecutor.getQueue();
                            executorQueue.setCapacity(MapUtils.getInteger(properties, "thread.pool.queueSize", queueSize));
                            //允许的空闲时间，当超过了核心线程数之外的线程在空闲时间到达之后会被销毁
                            threadPoolExecutor.setKeepAliveTime(MapUtils.getInteger(properties, "thread.pool.keepalive", keepalive), TimeUnit.MILLISECONDS);
                            log.info("刷新线程池，核心线程数:{},线程的最大数量：{},缓存队列:{},空闲时间:{}ms,线程名称:{}", threadPoolExecutor.getCorePoolSize(), threadPoolExecutor.getMaximumPoolSize(), ((DyLinkedBlockingQueue)threadPoolExecutor.getQueue()).getCapacity(), threadPoolExecutor.getKeepAliveTime(TimeUnit.MILLISECONDS), taskExecutor.getThreadNamePrefix());
                        }
                    }
                });
            }
        }
    }
}
